package com.zzmm.socket.qq.server;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import com.zzmm.socket.qq.common.*;
import lombok.SneakyThrows;

import javax.security.sasl.AuthenticationException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;

import static com.zzmm.socket.qq.common.QQConstant.MessageType.*;
import static com.zzmm.socket.qq.common.QQConstant.ResponseStatus.SUCCESS;

/**
 * 客户端线程池
 */
public class ServerSocketManager {

    private static final Map<String, UserSocketManager> OLINE_USER_SOCKET = new ConcurrentHashMap<>();

    public static Map<String, UserSocketManager> listOnlineUserSocket() {
        return OLINE_USER_SOCKET;
    }

    /**
     * 获取用户socket管理器
     */
    public static UserSocketManager get(String userName) {
        return OLINE_USER_SOCKET.get(userName);
    }

    /**
     * 添加用户socket
     */
    public static void add(Socket socket) {
        try {
            UserSocketManager userSocketManager = new UserSocketManager(socket);
            OLINE_USER_SOCKET.put(userSocketManager.getUsername(), userSocketManager);
        } catch (Exception e) {
            ConsoleUtil.printf("客户端连接失败：%s\n", e.getMessage());
        }
    }

    /**
     * 用户下线
     */
    public static void remove(String userName) {
        OLINE_USER_SOCKET.remove(userName);
    }


    /**
     * 用户Socket线程
     */
    public static class UserSocketManager {

        private final Socket socket;
        private final User user;
        private final ObjectInputStream inputStream;
        private final ObjectOutputStream outputStream;
        private final ExecutorService writeTread;
        private final Queue<Message> messageQueue;

        @SneakyThrows
        public UserSocketManager(final Socket socket) {
            this.socket = socket;
            this.outputStream = new ObjectOutputStream(socket.getOutputStream());
            // ObjectInputStream创建时会读取数据，所以要确保对方有写入数据
            this.inputStream = new ObjectInputStream(socket.getInputStream());
            this.writeTread = Executors.newSingleThreadExecutor();

            if ((this.user = auth()) == null) {
                throw new AuthenticationException("用户认证失败！");
            }

            this.messageQueue = new LinkedList<>();
            new ReadThread().start();
            new HandlerThread().start();
        }

        @SneakyThrows
        private User auth() {
            Message message = (Message) inputStream.readObject();
            User user = (User) message.getContent();

            User dbUser = DataBase.getUserByUsername(user.getUsername());
            if (ObjectUtil.isNotNull(dbUser) && ObjectUtil.equals(user.getPassword(), dbUser.getPassword())) {
                // 登录成功
                ConsoleUtil.printf("用户【%s】上线啦\n", user.getUsername());

                message.setResponseStatus(SUCCESS);
                writeMessage(message);

                // 推送离线消息
                Queue<Message> messages = DataBase.listOfflineMessageByUsername(user.getUsername());
                while (CollUtil.isNotEmpty(messages)) {
                    writeMessage(messages.poll());
                }

            } else {
                // 登录失败
                message = Message.builder()
                        .content("登录失败！用户名或密码错误")
                        .build();
                outputStream.writeObject(message);
                socket.close();
                return null;

            }

            return user;
        }

        /**
         * 下线
         */
        @SneakyThrows
        public boolean handlerLogout(Message message) {
            if (!LOGOUT.equals(message.getMessageType())) {
                return false;
            }

            // 通知客户端退出
            message.setResponseStatus(SUCCESS);
            outputStream.writeObject(message);

            // 释放资源
            socket.close();
            // 释放线程资源
            writeTread.shutdown();
            remove(getUsername());
            ConsoleUtil.printf("用户【%s】退出系统\n", user.getUsername());

            return true;
        }


        /**
         * 写线程
         */
        @SneakyThrows
        public void writeMessage(Message message) {
            if (socket.isClosed()) {
                return;
            }

            writeTread.submit(() -> {
                try {
                    outputStream.writeObject(message);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }

        /**
         * 处理线程
         */
        private final class HandlerThread extends Thread {

            private final Map<Integer, Consumer<Message>> HANDLER = new HashMap<>();

            {
                HANDLER.put(LIST_ONLINE_USER, this::sendListOnlineUsernames);
                HANDLER.put(PRIVATE_CHAT, this::handlerPrivateChat);
                HANDLER.put(GROUP_CHAT, this::handlerGroupChat);
                HANDLER.put(FILE, this::handlerFile);
            }

            private final Consumer<Message> DEFAULT_HANDLER = this::defaultHandler;

            private void defaultHandler(Message message) {
                ConsoleUtil.printf("用户【%s】发送了未知的消息：%s\n", getUsername(), message);
            }

            @SneakyThrows
            @Override
            public void run() {

                while (!socket.isClosed()) {
                    Thread.sleep(1000); // 本地调试减轻机器压力
                    Message message = messageQueue.poll();
                    if (ObjectUtil.isNull(message)) {
                        continue;
                    }

                    // 处理消息
                    HANDLER.getOrDefault(message.getMessageType(), DEFAULT_HANDLER).accept(message);

                }

                ConsoleUtil.printf("用户【%s】HandlerThread close\n", getUsername());
            }

            /**
             * 处理私聊
             */
            private void handlerPrivateChat(Message message) {
                // 设置发送人
                message.setSender(getUsername());

                UserSocketManager userSocketManager = get(message.getReceiver());
                if (ObjectUtil.isNull(userSocketManager)) {
                    DataBase.saveOfflineMessageByUsernameIfExists(message);
                    return;
                }


                // 转发私聊消息
                userSocketManager.writeMessage(message);
            }

            /**
             * 处理群聊
             */
            private void handlerGroupChat(Message message) {
                message.setSender(getUsername());

                DataBase.listAllUsername().stream()
                        .filter(username -> ObjectUtil.notEqual(username, getUsername()))
                        .forEach(username -> {
                            // 设置接收人
                            message.setReceiver(username);

                            UserSocketManager userSocketManager = OLINE_USER_SOCKET.get(username);
                            if (ObjectUtil.isNull(userSocketManager)) {
                                // 离线存储
                                DataBase.saveOfflineMessageByUsernameIfExists(message);
                            } else {
                                userSocketManager.writeMessage(message);
                            }
                        });
            }


            /**
             * 发送在线用户列表
             */
            private void sendListOnlineUsernames(Message message) {
                List<String> listOnlineUsernames = new ArrayList<>(OLINE_USER_SOCKET.keySet());
                message = Message.builder()
                        .content(listOnlineUsernames)
                        .messageType(QQConstant.MessageType.LIST_ONLINE_USER)
                        .build();
                writeMessage(message);
            }

            /**
             * 处理文件
             */
            private void handlerFile(Message message) {
                message.setSender(getUsername());

                UserSocketManager userSocketManager = OLINE_USER_SOCKET.get(message.getReceiver());
                if (ObjectUtil.isNull(userSocketManager)) {
                    DataBase.saveOfflineMessageByUsernameIfExists(message);
                    return;
                }

                userSocketManager.writeMessage(message);
            }


        }

        /**
         * 读线程
         */
        private final class ReadThread extends Thread {

            @SneakyThrows
            @Override
            public void run() {

                while (!socket.isClosed()) {
                    ConsoleUtil.println("服务器正在等待客户端消息...");

                    Message message = (Message) inputStream.readObject();

                    if (handlerLogout(message)) {
                        break;
                    }

                    messageQueue.offer(message);
                }

                ConsoleUtil.printf("用户【%s】ReadThread close\n", getUsername());

            }

        }

        /**
         * 获取用户名
         */
        public String getUsername() {
            return user.getUsername();
        }

    }

}
